#pragma once 

#include <vector>
#include <iostream>
#include <cstdint>
#include <cassert>
#include <cstring>
#include <cstdint>
#include <string>
#include <cstring>
#include <unistd.h>
#include <sys/types.h>        
#include <sys/socket.h>
#include <arpa/inet.h>
#include <fcntl.h>
#include <ctime>
#include <cstdint>
#include <iostream>
#include <functional>
#include <sys/epoll.h>
#include <sys/epoll.h>
#include <unordered_map>
#include <vector>
#include <unistd.h>
#include <cstdlib>
#include <cstring>
#include <cassert>
#include <thread>
#include <mutex>
#include <stdarg.h>
#include <sys/eventfd.h>
#include <sys/timerfd.h>
#include <memory>
#include <iostream>
#include <string>
#include <cassert>
#include <typeinfo>
#include <typeindex>
#include <unistd.h>
#include <mutex>
#include <thread>
#include <condition_variable>

#define SIZE 1024

#define Info 0
#define Debug 1
#define Warning 2
#define Error 3
#define Fatal 4

#define Screen 1    // 往屏幕打印
#define Onefile 2   // 往一个文件打印
#define Classfile 3 // 分文件打印

const std::string filename = "log.txt";

std::string LevelToTtring(int level)
{
    switch (level)
    {
    case Info:
        return "Info";
    case Debug:
        return "Debug";
    case Warning:
        return "Warning";
    case Error:
        return "Error";
    case Fatal:
        return "Fatal";
    default:
        return "None";
    }
}

class Log
{
public:
    Log()
    {
        printMethod = Screen; // 默认往屏幕打印
        path = "./log/";
    }

    void Enable(int method)
    {
        printMethod = method;
    }

    void PrintLog(int level, const std::string &logtxt)
    {
        switch (printMethod)
        {
        case Screen:
            std::cout << logtxt;
            break;
        case Onefile:
            printOneFile(filename, logtxt);
            break;
        case Classfile:
            printClassFile(level, logtxt);
            break;
        default:
            break;
        }
    }

    void printOneFile(const std::string &filename, const std::string &logtxt)
    {
        std::string filename_=path+filename;
        int fd = open(filename_.c_str(), O_APPEND | O_CREAT | O_WRONLY, 0666);
        if (fd < 0)
            return;
        write(fd, logtxt.c_str(), logtxt.size());
    }

    void printClassFile(int level, const std::string &logtxt)
    {
        std::string file = filename + "." + LevelToTtring(level);
        printOneFile(file, logtxt);
    }

    void operator()(int level, const char *format, ...)
    {
        // 获取本机时间
        time_t t = time(NULL);
        struct tm *ctime = localtime(&t);

        // 日志等级和时间
        char leftbuffer[SIZE] = {0};
        snprintf(leftbuffer, sizeof(leftbuffer), "[%s][%p][%d-%d-%d %d-%d-%d]", LevelToTtring(level).c_str(),
                (void*)pthread_self(),
                 ctime->tm_year + 1900, ctime->tm_mon + 1, ctime->tm_mday, ctime->tm_hour, ctime->tm_min, ctime->tm_sec
                 );

        // 日志信息
        char rightbuffer[SIZE] = {0};
        va_list s;
        va_start(s, format); // 让s指向可变参数列表的第一个参数
        vsnprintf(rightbuffer, sizeof(rightbuffer), format, s);
        va_end(s);

        // 合并
        char logbuffer[SIZE * 3] = {0};
        snprintf(logbuffer, sizeof(logbuffer), "%s %s\n", leftbuffer, rightbuffer);

        PrintLog(level, logbuffer);
    }

private:
    int printMethod; // 打印方法
    std::string path;     // 往文件打印的文件路径
};

Log lg;


class Buffer
{
    static const int defaultBufferSize = 1024; //缓冲区默认大小
private:
    //使用vector来充当缓冲区存放数据，为什么不使用string,应为string的很多操作都是争对字符串的，他遇到\0就会结束。而网络中会遇到各种各样的数据。
    std::vector<char> _buffer;
    //当前读位置偏移量 
    uint64_t _read_index;
    //当前写位置偏移量 
    uint64_t _write_index;
public:
    Buffer():_buffer(defaultBufferSize),_read_index(0),_write_index(0){}
private:
    //begin会返回一个迭代器，*对他解引用，&取出_buffer中首元素地址
    char *Begin() {return &*_buffer.begin();}
public:
    //获取当前读位置地址
    char* ReadPosition() {return Begin() + _read_index;}
    //获取当前写位置地址
    char* WritePosition() {return Begin() + _write_index;}
    //获取缓冲区可读数据大小
    uint64_t ReadAbleSize() {return _write_index - _read_index;}
    //获取缓冲区头部剩余空间大小
    uint64_t HeadIdleSize() {return _read_index;}
    //获取缓冲区尾部剩余空间大小
    uint64_t TailIdleSize() {return _buffer.size() - _write_index;}
    //确保可写空间⾜够（整体空闲空间够了就移动数据，否则就扩容）
    void EnsureWriteSpace(uint64_t len)
    {
        //如果尾部空间足够就直接返回
        if(len <= TailIdleSize())
        {
            return;
        }
        //尾部空间不够，但是尾部空间+头部空间足够，那么就需要挪动数据至起始位置
        else if(len <= TailIdleSize() + HeadIdleSize())
        {
            uint64_t size = ReadAbleSize();
            //将数据挪动到起始位置
            std::copy(ReadPosition(),WritePosition(),Begin());
            _read_index = 0;
            _write_index = size;
        }
        //直接在尾部扩容，扩容到刚好放下len个数据
        else
        {
            _buffer.resize(_write_index + len);
        }
    }
    //移动读偏移
    void MoveReadOffset(uint64_t len)
    {
        //移动的距离必须小于可读数据大小
        assert(len <= ReadAbleSize());
        _read_index += len;
    }
    //移动写偏移
    void MoveWriteOffset(uint64_t len)
    {
        //移动的距离必须小于等于缓冲区尾部剩余空间大小
        assert(len <= TailIdleSize());
        _write_index +=len;
    }

    //清空缓冲区
    void clear()
    {
        _read_index = 0;
        _write_index = 0;
    }

public:
    //写入数据
    void Write(const void *data,uint64_t len)
    {
        if(len == 0) return;
        EnsureWriteSpace(len);

        //copy是一个模板函数，三个参数的类型必须一致
        const char* d = (const char*)data;
        std::copy(d,d + len,WritePosition());
    }

    void WriteAndPush(const void *data,uint64_t len)
    {
        if(len == 0) return ;
        Write(data,len);
        MoveWriteOffset(len);
    }

    //把一个Buffer内的数据写入BUffer中
    void WriteBuffer(Buffer &buf)
    {
        Write(buf.ReadPosition(),buf.ReadAbleSize());
    }

    void WriteBufferAndPush(Buffer &buf)
    {
        WriteBuffer(buf);
        MoveWriteOffset(buf.ReadAbleSize());
    }

    //把一个string写入Buffer中
    void WriteAsString(const std::string &str)
    {
        Write(str.c_str(),str.size());
    }

    void WriteAsStringAndPush(const std::string &str)
    {
        WriteAsString(str);
        MoveWriteOffset(str.size());
    }

public:
    //读取数据
    void Read(void *data,uint64_t len)
    {
        //必须保证获取的数据⼤⼩⼩于可读数据⼤⼩
        assert(len <= ReadAbleSize());
        std::copy(ReadPosition(),ReadPosition() + len,(char*)data);
    }

    void ReadAndPop(void *data,uint64_t len)
    {
        Read(data,len);
        MoveReadOffset(len);
    }

    std::string ReadAsString(uint64_t len)
    {
        //必须保证获取的数据⼤⼩⼩于可读数据⼤⼩
        assert(len <= ReadAbleSize());
        std::string ret;
        ret.resize(len);
        Read(&ret[0],len);
        return ret;
    }

    std::string ReadAsStringAndPop(uint64_t len)
    {
        //必须保证获取的数据⼤⼩⼩于可读数据⼤⼩
        assert(len <= ReadAbleSize());
        std::string ret = ReadAsString(len);
        MoveReadOffset(len);
        return ret;
    }
public:
//为HTTP服务提供的接口

    //寻找回车换行字符所在的位置
    char* FindCRLF()
    {
        //找寻一个字符在一段内存中的位置
        char *pos = (char*)memchr(ReadPosition(),'\n',ReadAbleSize());
        return pos;
    }

    //获取一行，把'\n'字符也获取到
    std::string GetLine()
    {
        char *pos = FindCRLF();
        if(pos == nullptr)
        {
            return "";
        }

        //这里+1的目的是为了把'\n'字符也提取出来
        return  ReadAsString(pos - ReadPosition() + 1);
    }

    std::string GetLineAndPop()
    {
        std::string ret = GetLine();
        MoveReadOffset(ret.size());
        return ret;
    }
};


class Socket
{
    static const int max_backlog = 1024;    //最大连接队列长度
private:
    int sockfd_;
public:
    Socket():sockfd_(-1){}
    Socket(int fd):sockfd_(fd){}
    ~Socket(){Close();}

    int Fd(){return sockfd_;}
public:
    //创建套接字
    bool Create()
    {
        sockfd_ = socket(AF_INET,SOCK_STREAM,0);
        if(sockfd_ < 0){
            lg(Fatal,"create socket failed");
            return false;
        }
        return true;
    }
    //绑定ip端口号
    bool Bind(uint16_t port,std::string ip = "0.0.0.0")
    {
        struct sockaddr_in local;
        memset(&local,0,sizeof(local));
        local.sin_family = AF_INET;
        local.sin_port = htons(port);
        local.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len =sizeof(local);

        int ret = bind(sockfd_,(const struct sockaddr*)&local,len);
        if(ret < 0){
            lg(Fatal,"server bind failed");
            return false;
        }

        return true;
    }
    //开始监听
    bool Listen(int backlog = max_backlog)
    {
        int ret = listen(sockfd_,backlog);
        if(ret < 0)
        {
            lg(Fatal,"server listen failed");
            return false;
        }

        return true;
    }
    //获取新连接
    int Accept(struct sockaddr_in *client = nullptr)
    {   
        socklen_t len = sizeof(sockaddr_in);
        int newfd = accept(sockfd_, (struct sockaddr*)client, &len);
        if (newfd < 0) {
            lg(Fatal,"SOCKET ACCEPT FAILED!");
            return -1;
        }
        return newfd;
    }
    //向服务器发起连接
    bool Connect(uint16_t port,std::string ip)
    {
        struct sockaddr_in server;
        memset(&server,0,sizeof(server));
        server.sin_family = AF_INET;
        server.sin_port = htons(port);
        server.sin_addr.s_addr = inet_addr(ip.c_str());
        socklen_t len =sizeof(server);

        int ret = connect(sockfd_,(const struct sockaddr*)&server,len);
        if(ret < 0){
            lg(Fatal,"client connect failed");
            return false;
        }

        return true;
    }
    //发送数据
    ssize_t Send(const void *data,size_t len,int flag = 0)
    {
        ssize_t ret = send(sockfd_,data,len,flag);
        //发送出错了
        if(ret < 0)
        {
            //发送缓冲区满了，或者阻塞在send的时候被信号打断了
            //这种错误是可以被原谅的，我们return 0
            if(errno == EAGAIN || errno == EINTR)
            {
                return 0;
            }
            //其他情况就是真的出错了
            lg(Error,"send failed");
            return -1;
        }

        //返回实际发送的数据大小
        //data中的数据是有可能没有全部发送完的
        return ret;
    }
    ssize_t SendNonBlock(void *data,size_t len)
    {
        return Send(data,len,MSG_DONTWAIT);
    }

    //接收数据
    ssize_t Recv(void *data,size_t len,int flag = 0)
    {
        int ret = recv(sockfd_,data,len,flag);
        //==0代表关闭了连接也是出错了
        if(ret <= 0){

            if(errno == EAGAIN || errno == EINTR){
                return 0;
            }

            //真正出错的情况
            lg(Error,"recv failed, err string:%s",strerror(errno));
            return -1;
        }

        //实际接收的数据长度
        return ret;
    }

    ssize_t RecvNonBlock(void *data,size_t len)
    {
        return Recv(data,len,MSG_DONTWAIT);
    }

    //创建一个服务器连接
    bool CreateServer(uint16_t port,const std::string &ip = "0.0.0.0",bool BlockFalg = false)
    {
        if(Create() == false) return false;
        //设置套接字非阻塞
        if(BlockFalg) {
            NonBlock();
        }
        if(Bind(port,ip) == false) return false;
        if(Listen(sockfd_) == false) return false;
        //设置地址端口复用
        ReuseAddr();
        return true;
    }
    //创建一个客户端连接
    bool CreateClient(uint16_t port,const std::string &ip)
    {
        if(Create() == false) return false;
        if(Connect(port,ip) == false) return false;

        return true;
    }
    //开启地址端口复用
    void ReuseAddr()
    {
        int optval = 1;
        if(setsockopt(sockfd_, SOL_SOCKET, SO_REUSEADDR, &optval, sizeof(optval)) < 0){
            lg(Fatal,"setsockopt(SO_REUSEADDR) failed");
        }

        optval = 1;
        if(setsockopt(sockfd_, SOL_SOCKET, SO_REUSEPORT, &optval, sizeof(optval)) < 0){
            lg(Fatal,"setsockopt(SO_REUSEPORT) failed");
        }
    }
    //设置描述符非阻塞
    void NonBlock()
    {
        int flag = fcntl(sockfd_,F_GETFL,0);
        fcntl(sockfd_,F_SETFL, flag | O_NONBLOCK);
    }
    //关闭套接字
    void Close(){
        if(sockfd_ != -1) {
            close(sockfd_);
        }
    }
};

class EventLoop;
class Poller;
class Channel
{
private:
    int fd_;
    uint32_t events_; //描述符需要监控的事件
    uint32_t revents_; //描述符已经触发的事件

    EventLoop *loop_; //绑定的EventLoop

    using EventCallback = std::function<void()>;
    EventCallback read_callback_;  //可读事件触发执行的回调函数
    EventCallback write_callback_; //可写事件触发执行的回调函数
    EventCallback close_callback_; //连接断开事件触发执行的回调函数
    EventCallback error_callback_; //错误事件触发执行的回调函数
    EventCallback event_callback_; //任意事件触发执行的回调函数
public:
    Channel(EventLoop* loop,int fd):loop_(loop),fd_(fd),events_(0),revents_(0){}
    int Fd() {return fd_;}
    uint32_t Events(){return events_;}
    void SetREvents(uint32_t revents) { revents_ = revents;}
    void SetReadCallback(const EventCallback &cb){ read_callback_ = cb; }
    void SetWriteCallback(const EventCallback &cb){ write_callback_ = cb; }
    void SetCloseCallback(const EventCallback &cb){ close_callback_ = cb; }
    void SetErrorCallback(const EventCallback &cb){ error_callback_ = cb; }
    void SetEventCallback(const EventCallback &cb){ event_callback_ = cb; }
public:
    //是否监控了可读事件
    bool ReadAble() {return (events_ & EPOLLIN); }
    //是否监控了可写事件
    bool WriteAble() {return (events_ & EPOLLOUT); }
    //监控可读事件
    void EnableRead() {events_ |= EPOLLIN ; Update();}
    //监控可写事件
    void EnableWrite() {events_ |= EPOLLOUT ;Update();}
    //移除可读事件监控
    void DisAbleRead() {events_ &= (~EPOLLIN); Update();}
    //移除可写事件监控
    void DisAbleWrite() {events_ &= (~EPOLLOUT); Update();}
    //移除所有事件监控
    void DisAbleAll() {events_ = 0; Update();}
    //解除监控
    void Remove() ;
    void Update() ;
    //事件处理  EventLoop不关心事件触发后怎么处理，事件触发后，它只需要调用HandleEvent函数，这个函数内触发了什么事件后执行什么事件触发后的回调
    void HandleEvent()
    {   
        //EPOLLRDHUP代表对端关闭了连接，我们也调用可读事件回调，把缓冲区中的数据发送，但是对端不会收到信息
        //EPOLLPRI代表有紧急数据，我们也调用可读事件回调，读取数据，进行处理

        if((revents_ & EPOLLIN) || (revents_ & EPOLLRDHUP) || (revents_ & EPOLLPRI))
        {
              //调用任意事件回调
            if(event_callback_) event_callback_();
            if(read_callback_) read_callback_();
        }

        //对于可能释放连接的操作，我们只执行一次
        //在进行可写事件回调时，有可能连接关闭事件触发，我们就要释放连接，如果还执行后面的close_callback_ 和 error_callback_就会出问题
        if(revents_ & EPOLLOUT)
        {
            //调用任意事件回调
            if(event_callback_) event_callback_();
            if(write_callback_) write_callback_();
        }
        else if(revents_ & EPOLLHUP)
        {
            if(close_callback_) close_callback_();
        }
        else if(revents_ & EPOLLERR)
        {
            if(error_callback_) error_callback_();
        }

    }
};


class Channel;
class Poller
{
    static const int max_events = 1024;
private:
    int epfd_;
    struct epoll_event evs[max_events]; //存放活跃连接的数组
    std::unordered_map<int,Channel*> channels_; //添加到epoll监控中的Channel <文件描述符，和Channel*>
public:
    Poller()
    {
        epfd_ = epoll_create(max_events);
        if(epfd_ < 0){
            lg(Fatal,"epoll create failed");
            abort();    //终止程序
        }
    }
private:
    void Update(Channel *channel,int op)
    {
        struct epoll_event ev;
        ev.data.fd = channel->Fd();
        ev.events = channel->Events();
        int ret = epoll_ctl(epfd_,op,channel->Fd(),&ev);
        if(ret < 0){
            lg(Error,"epoll ctl failed,err string: %s",strerror(errno));
        }
        return ;

    }
    bool HasChannels(Channel *channel)
    {
        auto it = channels_.find(channel->Fd());
        if(it == channels_.end()){
            return false;
        }
        return true;
    }
public:
    //添加或修改监控
    void UpdateEvent(Channel *channel)
    {
        bool ret = HasChannels(channel);
        if(ret == false){
            //不存在，则添加
            channels_.insert({channel->Fd(),channel});
            return Update(channel,EPOLL_CTL_ADD);
        }
        //存在，则修改
        return Update(channel,EPOLL_CTL_MOD);
    }
    //移除监控
    void RemoveEvent(Channel* channel)
    {
        //从哈希表中移除
        auto it = channels_.find(channel->Fd());
        if(it != channels_.end()){
            channels_.erase(it);
        }
        //从Epoll中移除
        Update(channel,EPOLL_CTL_DEL);


    }
    //开始监控，并返回活跃连接
    void Poll(std::vector<Channel*> *actives)
    {
        //-1代表阻塞等待
        memset(evs,0,sizeof(evs));
        int nfds = epoll_wait(epfd_,evs,max_events,-1);
        if(nfds < 0){
            lg(Error,"epoll wait failed,err string:%s",strerror(errno));
            abort(); //终止程序
        }

        for(int i = 0; i < nfds; i++){
            auto it = channels_.find(evs[i].data.fd);
            //这个channel一定要存在哈希表中
            assert(it != channels_.end());
            //给channel设置触发了的事件
            it->second->SetREvents(evs[i].events);
            actives->push_back(it->second);
        }

    }
};


using TaskFunc = std::function<void()>;
using ReleaseFunc = std::function<void()>;
class TimerTask
{
private:
    uint64_t timer_id_; // 定时任务ID
    uint32_t timeout_; //任务执行时间，也就是多少秒后执行
    bool canceld_; //是否取消定时任务，false代表不取消，true代表取消
    TaskFunc task_cb; //真正的要执行的任务
    ReleaseFunc release_cb; //用于删除TimerWheer中哈希表中的信息
public:
    TimerTask(uint64_t timer_id,uint32_t delay,const TaskFunc &cb):
        timer_id_(timer_id),timeout_(delay),canceld_(false),task_cb(cb){}

    //取消定时任务
    void SetCancle(){ canceld_ = true; }
    void SetRelease(const ReleaseFunc &cb){ release_cb = cb; }
    //返回任务的延迟时间，在TimerWheel中给延迟任务的接口使用
    uint32_t Delay(){ return timeout_; }

    ~TimerTask(){
        if(canceld_ == false){
            task_cb();
        }

        release_cb();
    }
};

class EventLoop;
class TimerWheel
{
private:
    using PtrTask = std::shared_ptr<TimerTask>;
    using WeakTask = std::weak_ptr<TimerTask>;
    int tick_; //秒针，走到哪里就释放哪里，相当于执行哪里的定时任务
    int capcity_; //时间轮的大小，就相当于定时任务的最大延迟时间
    //因为我们的秒针是要一秒往后移动一位的，所以我们需要有timerfd来定位一秒
    //监控timerfd的可读事件，并将可读事件的回调设置为读取timefd,并执行RunTimerTask
    EventLoop *loop_; //我们的timerfd_channel_需要挂接到一个EventLoop上
    int timerfd_;   //定时器的fd
    std::unique_ptr<Channel> timerfd_channel_; //对定时器描述符事件管理的channel
    std::vector<std::vector<PtrTask>> wheel_; //时间轮
    std::unordered_map<uint64_t,WeakTask> timers_; //存放所有的定时任务的weak_ptr,用于延迟定时任务操作
public:
    TimerWheel(EventLoop *loop)
        :tick_(0),capcity_(60),wheel_(capcity_),
        loop_(loop),
        timerfd_(CreateTimerFd()),
        timerfd_channel_(new Channel(loop_,timerfd_))
    {
        //设置timerfd读事件回调
        timerfd_channel_->SetReadCallback(std::bind(&TimerWheel::OnTime,this,timerfd_));
        //开启读监控
        timerfd_channel_->EnableRead();
    }
private:
    //从哈希表中移除定时任务信息
    void RemoveTimer(uint64_t timer_id)
    {
        auto it = timers_.find(timer_id);
        if(it == timers_.end()){
            //没找到，不能删除
            return;
        }

        timers_.erase(it);
    }

    //创建定时器并启动 ,这个函数就会每秒钟触发一次timerfd_channel的可读事件
    static int CreateTimerFd()
    {
        int timerfd = timerfd_create(CLOCK_MONOTONIC,0);
        if(timerfd < 0){
            lg(Fatal,"timerfd create failed");
            abort();
        }

        //启动定时器
        struct itimerspec timer;
        timer.it_value.tv_sec = 1;
        timer.it_value.tv_nsec = 0;
        timer.it_interval.tv_sec = 1;
        timer.it_interval.tv_nsec = 0;
        int n = timerfd_settime(timerfd,0,&timer,nullptr);
        if(n < 0){
            lg(Fatal,"timer start failed");
            abort();
        }

        return timerfd;
    }

   
    uint64_t ReadTimerFd(int timerfd)
    {
        uint64_t val = 0;
        int n = read(timerfd,&val,sizeof(val));
        if(n < 0){
            lg(Fatal,"read timerfd failed");
            abort();
        }

        return val;
    }

    //这个函数就是timerfd_channel的可读事件回调
    void OnTime(int timerfd)
    {
        //读取超时次数
        uint64_t count = ReadTimerFd(timerfd);
        //执行定时任务
        for(int i = 0; i < count; i++)
            RunTimerTask();
    }

      //添加定时任务
    void TimerAddInLoop(uint64_t timer_id,uint32_t delay,const TaskFunc &cb){
        PtrTask pt(new TimerTask(timer_id,delay,cb));

        pt->SetRelease(std::bind(&TimerWheel::RemoveTimer,this,timer_id));
        int pos = (tick_ + delay) % capcity_;
        wheel_[pos].push_back(pt);
        timers_[timer_id] = WeakTask(pt);
    }

    //刷新/延迟定时任务
    void TimerRefreshInLoop(uint64_t timer_id){
        auto it = timers_.find(timer_id);
        if(it == timers_.end()){
            return;
        }
        //构建一个shared_ptr
        PtrTask pt = it->second.lock();
        uint32_t delay = pt->Delay();
        int pos = (tick_ + delay) % capcity_;
        wheel_[pos].push_back(pt);
    }
    //取消定时任务
    void TimerCancelInLoop(uint64_t timer_id){
        auto it = timers_.find(timer_id);
        if(it == timers_.end()){
            return;
        }

        PtrTask pt = it->second.lock();
        if(pt) pt->SetCancle();
    }

    

public:
    //这三个函数，都涉及到了对Wheel_的操作，他是会有线程安全的，我们要么加锁，要么让他在一个线程中执行。我们选择压入EventLoop的任务队列中
    void TimerAdd(uint64_t timer_id,uint32_t delay,const TaskFunc &cb);
    void TimerRefresh(uint64_t timer_id);
    void TimerCancel(uint64_t timer_id);

    //执行定时任务，让秒针向后走一步，这个函数应该每秒钟向后走一步
    void RunTimerTask()
    {   
        tick_ = (tick_ + 1) % capcity_;
        wheel_[tick_].clear();
    }

    //这个函数是有线程安全问题的，我们只能在组件内由组件调用，不能给用户提供使用
    bool HasTimer(uint64_t timer_id){
        auto it = timers_.find(timer_id);
        if(it == timers_.end()){
            return false;
        }
        
        return true;
    }
};

class EventLoop
{
private:
    std::thread::id thread_id_; // EventLoop对象绑定线程的线程ID 用于判断连接的各种操作是否处在当前EventLoop绑定的线程
    Poller poller;
    int event_fd_;                           // 事件通知fd
    std::unique_ptr<Channel> event_channel_; // event_fd_对应的channel
    using Functor = std::function<void()>;
    std::vector<Functor> tasks_; // 任务池
    std::mutex mutex_;           // 维护任务池线程安全的锁
    TimerWheel wheel_;  
private:
    static int CreateEventFd()
    {
        // 创建事件通知fd
        // EFD_CLOEXEC代表文件描述符不会被子进程复制
        // EFD_NONBLOCK 代表非阻塞
        int event_fd = eventfd(0, EFD_CLOEXEC | EFD_NONBLOCK);
        if (event_fd < 0)
        {
            lg(Fatal,"create eventfd failed");
            abort();
        }
        return event_fd;
    }

    void ReadEventFd()
    {
        uint64_t val = 0;
        int n = read(event_fd_,&val,sizeof(val));
        if(n < 0){

            //被信号打断或者没有数据可读
            if(errno == EINTR || errno == EAGAIN){
                return ;
            }

            //否则即使真的出错了
            lg(Fatal,"read eventfd error");
            abort();
        }
    }

    void WeakUpEventFd()
    {
        uint64_t val = 1;
        int n = write(event_fd_,&val,sizeof(val));
        if(n < 0){
            if(errno == EINTR){
                return;
            }

            lg(Fatal,"write eventfd error");
            abort();
        }
    }

    void RunAllTask(){
        std::vector<Functor> functor;
        {
            //对任务池的操作是有线程安全的，所以加锁
            std::unique_lock<std::mutex> lock(mutex_);
            tasks_.swap(functor);
        }

        //执行任务池的任务没有线程安全，应为执行的时候一定是当前线程中
        for(auto &f:functor){
            f();
        }

    }

public:
    EventLoop()
        : thread_id_(std::this_thread::get_id()),
          event_fd_(CreateEventFd()),
          event_channel_(new Channel(this,event_fd_)),
          wheel_(this)
    {
        //设置event_channel的读事件回调
        event_channel_->SetReadCallback(std::bind(&EventLoop::ReadEventFd,this));
        //启动读监控
        event_channel_->EnableRead();
    }

public:
    //切换协议---重置上下⽂以及阶段性回调处理函数 -- ⽽是这个接⼝必须在EventLoop线程中 “⽴即执⾏”
    //防备新的事件触发后，处理的时候，切换任务还没有被执⾏--会导致数据使⽤原协议处理了。
    void AssertInLoop()
    {
        assert(thread_id_ == std::this_thread::get_id());
    }

    bool IsInLoop(){
         return thread_id_ == std::this_thread::get_id();
    }

    // 如果当前连接的操作处在当前对应线程，就直接执行， 否则压入任务池
    void RunInLoop(const Functor &cb){

        if(IsInLoop()){
            cb();
        }

        //压入任务池
        else QueueInLoop(cb);
    }
    // 将连接的操作压入任务池
    void QueueInLoop(const Functor &cb){
        {
            std::unique_lock<std::mutex> lock(mutex_);
            tasks_.push_back(cb);
        }

        //唤醒可能因为没有事件就绪，而导致epoll阻塞,我们的任务得不到执行
        WeakUpEventFd();
    }
    // 开始运行
    void start(){
        while(1)
        {
            //1.事件监控，返回活跃链接
            std::vector<Channel*> actives;
            poller.Poll(&actives);
            //2.执行对应触发的事件的回调函数
            for(auto &a:actives){
                a->HandleEvent();
            }
            //3.执行任务队列的任务
            RunAllTask();
        }
        
    }
    // 添加或修改监控
    void UpdateEvent(Channel *channel){
        poller.UpdateEvent(channel);
    }
    // 移除监控
    void RemoveEvent(Channel *channel){
        poller.RemoveEvent(channel);
    }

    void TimerAdd(uint64_t timer_id,uint32_t delay,const TaskFunc &cb){
         wheel_.TimerAdd(timer_id,delay,cb); 
    }

    void TimerRefresh(uint64_t timer_id){
        wheel_.TimerRefresh(timer_id);
    }

    void TimerCancel(uint64_t timer_id){
        wheel_.TimerCancel(timer_id);
    }

     bool HasTimer(uint64_t timer_id){
        return wheel_.HasTimer(timer_id);
     }
};

void Channel::Remove() {loop_->RemoveEvent(this);}
void Channel::Update() {loop_->UpdateEvent(this);}

//这三个函数，都涉及到了对Wheel_的操作，他是会有线程安全的，我们要么加锁，要么让他在一个线程中执行。我们选择压入EventLoop的任务队列中
void TimerWheel::TimerAdd(uint64_t timer_id,uint32_t delay,const TaskFunc &cb){
    loop_->RunInLoop(std::bind(&TimerWheel::TimerAddInLoop,this,timer_id,delay,cb));
}
void TimerWheel::TimerRefresh(uint64_t timer_id){
    loop_->RunInLoop(std::bind(&TimerWheel::TimerRefreshInLoop,this,timer_id));
}
void TimerWheel::TimerCancel(uint64_t timer_id){
    loop_->RunInLoop(std::bind(&TimerWheel::TimerCancelInLoop,this,timer_id));
}



class Any
{
    class holder
    {
    public:
        virtual ~holder() {}  //父类指针指向子类，为了保证delete父类时，能正确释放子类
        virtual const std::type_info &type() = 0;
        virtual holder *clone() = 0;
    };

    template<class T>
    class placeholder : public holder
    {
    public:
        placeholder(const T& val):val_(val) {}
        ~placeholder() {}
        const std::type_info &type() { return typeid(T); }
        holder *clone() { return new placeholder(val_); }   //提供这个函数的目的，是为了方便我们通过一个Any对象来构建一个Any对象
    public:
        T val_;
    };

public:
    Any():content_(nullptr) {}

    template<class T>
    Any(const T &val) : content_(new placeholder<T>(val)){}

    Any(const Any &other){
        content_ = other.content_ ? other.content_->clone() : nullptr ;
    }

    ~Any() { delete content_; }

    template<class T>
    T *get(){
        //要获取的类型必须和实际存储的类型一致
        assert(content_->type() == typeid(T));
        return &((placeholder<T>*)content_)->val_;
    }

    template<class T>
    Any &operator=(const T&val){
        //构造一个临时的Any匿名对象
        Any(val).swap(*this);
        return *this;
    }

    Any &operator=(Any other){
        //这里的形参没有用&,会构造一个临时Any,出了作用域会自动销毁
        other.swap(*this);
        return *this;
    }

    //我们自己实现的swap,方便我们operator=操作
    Any& swap(Any &other){
        std::swap(other.content_,this->content_);
        return *this;
    }
private:
    holder *content_;   //父类指针，我们就是通过这个指针来对实际保存的数据操作的，通过多态
};

class Connection;
using PtrConnection = std::shared_ptr<Connection>; 
 
//连接的各种状态
enum connStatu{
    DISCONNECTED,  //连接关闭状态
    CONNECTING,    //连接建立成功，还需要进行各种回调设置
    CONNECTED,     //连接建立完成，可以进行通信
    DISCONNECTING  //连接半关闭状态
};
class Connection : public std::enable_shared_from_this<Connection>
{
private:
    uint64_t conn_id_;  //连接唯一ID，同时用这个参数代替timer_id
    EventLoop *loop_;   //connection绑定的一个EventLoop
    connStatu statu_;   //连接状态
    bool enable_inactive_release_; //连接是否启动非活跃超时销毁 默认为false
    int sockfd_;        //连接对应的描述符
    Channel channel_;   //连接描述符的channel

    struct sockaddr_in client_;  //连接所对应客户端的信息

    Socket socket_;     //进行套接字操作的socket
    Buffer in_buffer_;  //输入缓冲区
    Buffer out_buffer_; //输出缓冲区
    Any context_;   //上下文信息 客户端可能传送的报文不是一个完整的报文，我们需要把已经处理的报文信息及状态存储到这个Any对象中，等待客户端下一次传送剩下的报文。
    using ConnectedCallBack = std::function<void(const PtrConnection &)>;    
    using MessageCallBack = std::function<void(const PtrConnection &,Buffer *)>; 
    using ClosedCallBack = std::function<void(const PtrConnection &)>;   
    using AnyEventCallBack = std::function<void(const PtrConnection &)>;  
    using SrvClosedCallBack = std::function<void(const PtrConnection &)>; 

    ConnectedCallBack connected_callback_;              //连接建立回调
    MessageCallBack message_callback_;              //新消息到来回调
    ClosedCallBack closed_callback_;                //连接关闭回调
    AnyEventCallBack event_callback_;               //任意事件回调
    SrvClosedCallBack server_closed_callback_;      //组件内的连接关闭回调，因为组件内会把所有连接的信息管理起来，调用这个函数把连接的信息从组件内移除
    
public:
    Connection(EventLoop *loop,uint64_t conn_id,int sockfd,struct sockaddr_in client):
        conn_id_(conn_id),loop_(loop),statu_(CONNECTING),enable_inactive_release_(false),sockfd_(sockfd),channel_(loop_,sockfd_),socket_(sockfd),
        client_(client)
     {
        //设置channel回调函数
        //不能在构造函数中开启读事件监控，因为开启了读事件监控，可能立马就会有事件触发，那么就会触发任意事件，刷新活跃度，而我们此时还没有添加定时任务
        channel_.SetReadCallback(std::bind(&Connection::HandleRead,this));
        channel_.SetWriteCallback(std::bind(&Connection::HandleWrite,this));
        channel_.SetErrorCallback(std::bind(&Connection::HandleError,this));
        channel_.SetCloseCallback(std::bind(&Connection::HandleClose,this));
        channel_.SetEventCallback(std::bind(&Connection::HandleEvent,this));
     }

    //析构里打印一下日志，方便我们观察连接是否正常释放
     ~Connection() { lg(Debug,"release connection:%p",this); }
private:
    void HandleRead(){
        char buffer[65536];
        int n = socket_.RecvNonBlock(buffer,65535);
        if(n < 0){
            //recv出错了，或者对方关闭了连接的情况，不能直接关闭连接
            return ShutDown();
        }

        in_buffer_.WriteAndPush(buffer,n);  //将数据写入缓冲区
        if(in_buffer_.ReadAbleSize() > 0){
            return message_callback_(shared_from_this(), &in_buffer_);
        }
    }
    void HandleWrite(){
        int n = socket_.SendNonBlock(out_buffer_.ReadPosition(),out_buffer_.ReadAbleSize());
        if(n < 0){
            //这时候就是真的关闭了
            if(in_buffer_.ReadAbleSize() > 0){
                message_callback_(shared_from_this(),&in_buffer_);
            }

            return Release();
        }

        out_buffer_.MoveReadOffset(n);  //别忘了移动缓冲区读偏移
        if(out_buffer_.ReadAbleSize() == 0){
            channel_.DisAbleWrite(); //关闭写事件监控

            if(statu_ == DISCONNECTING){
                return Release();
            }
        }
    }
    void HandleClose(){
        //对端套接字已经关闭了，发送的信息对方也收不到，这是处理一下缓冲区中数据，关闭连接
        if(in_buffer_.ReadAbleSize() > 0){
            message_callback_(shared_from_this(),&in_buffer_);
        }

        return Release();
    }
    void HandleError(){
        HandleClose();
    }

    //任意事件触发，刷新活跃度，调用组件使用者设置的任意事件回调
    void HandleEvent(){
       if(enable_inactive_release_ == true){
            loop_->TimerRefresh(conn_id_);
       }

       if(event_callback_){
            event_callback_(shared_from_this());
       }
    }
     //发送数据,这个函数并不是真的发送，而是将要发送的信息放入输出缓冲区，然后开始写事件监控
    void SendInLoop(Buffer &buf,size_t size){
        //如果连接已经关闭了，则直接返回
        if(statu_ == DISCONNECTED) return;
        //将数据写入输出缓冲区
        out_buffer_.WriteAndPush(buf.ReadPosition(),size);
        //开启写事件监控
        if(channel_.WriteAble() == false)
            channel_.EnableWrite();
    }
    //关闭连接，这个函数并不是立马关闭连接，需要先看是否还有数据待发送，待处理。
    void ShutDownInLoop(){
        statu_ = DISCONNECTING;
        if(in_buffer_.ReadAbleSize() > 0){
            message_callback_(shared_from_this(),&in_buffer_);
        }

        if(out_buffer_.ReadAbleSize() > 0){
            if(channel_.WriteAble() == false){
                channel_.EnableWrite();
            }
        }

        if(out_buffer_.ReadAbleSize() == 0){
            Release();
        }
    }
    //关闭连接，这个函数才是真正的关闭连接
    void Release(){
        statu_ = DISCONNECTED;
        channel_.Remove();
        socket_.Close();
        if(loop_->HasTimer(conn_id_)){
            loop_->TimerCancel(conn_id_);
        }
        if(closed_callback_){
            closed_callback_(shared_from_this());
        }
        if(server_closed_callback_){
            
            server_closed_callback_(shared_from_this());
        }
    }
    //启动非活跃超时销毁
    void EnableInactiveReleaseInLoop(int sec){
        enable_inactive_release_ = true;   
        if(loop_->HasTimer(conn_id_)){
            loop_->TimerRefresh(conn_id_);
        }else{
            loop_->TimerAdd(conn_id_,sec,std::bind(&Connection::Release,this));
        }

    }
    //取消非活跃超时销毁
    void CancelInactiveareaseInLoop(){
        enable_inactive_release_ = false;
        if(loop_->HasTimer(conn_id_))
            loop_->TimerCancel(conn_id_);
    }

    void EstablishedInLoop(){
        //更改连接状态
        statu_ = CONNECTED;
        //启动读事件监控
        channel_.EnableRead();
        //调用回调函数
        if(connected_callback_)
            connected_callback_(shared_from_this());
    }

    //协议切换,就是把会回调函数和上下文切换掉
    void UpgradeInLoop(const Any &context,
                const ConnectedCallBack &connected,
                const MessageCallBack &message,
                const ClosedCallBack &closed,
                const AnyEventCallBack &event,
                const SrvClosedCallBack &srvclose)
    {
        context_ = context;
        connected_callback_ = connected;
        message_callback_ = message;
        closed_callback_ = closed;
        event_callback_ = event;
        server_closed_callback_ = srvclose;
    }

public:
    uint64_t Id() { return conn_id_; }
    int Fd() { return sockfd_; }
    void SetContext(const Any &context) { context_ = context; }
    Any *Context() { return &context_; }
    bool Connected() { return statu_ == CONNECTED; }
    void SetConnectedCallBack(const ConnectedCallBack &cb) { connected_callback_ = cb; }
    void SetMessageCallBack(const MessageCallBack &cb) { message_callback_ = cb; }
    void SetClosedCallBack(const ClosedCallBack &cb) {closed_callback_ = cb; }
    void SetAnyEventCallBack(const AnyEventCallBack &cb) { event_callback_ = cb; }
    void SetSrvClosedCallBack(const SrvClosedCallBack &cb) { server_closed_callback_ = cb; }
    //连接建立成功，开启读监控，调用用户设置的回调
    void Established(){
       loop_->RunInLoop(std::bind(&Connection::EstablishedInLoop,this));
    }
    //发送数据,这个函数并不是真的发送，而是将要发送的信息放入输出缓冲区，然后开始写事件监控
    void Send(void *data,size_t size){
        //因为send这个函数不是立马发送的，所以有可能外界要发送的信息会释放掉，这里我们保存一下
        Buffer buf;
        buf.WriteAndPush(data,size);
        loop_->RunInLoop(std::bind(&Connection::SendInLoop,this,std::move(buf),buf.ReadAbleSize()));
    }
    //关闭连接，这个函数并不是立马关闭连接，需要先看是否还有数据待发送，待处理。
    void ShutDown(){
         loop_->RunInLoop(std::bind(&Connection::ShutDownInLoop,this));
    }
    //启动非活跃超时销毁
    void EnableInactiveRelease(int sec){
        loop_->RunInLoop(std::bind(&Connection::EnableInactiveReleaseInLoop,this,sec));
    }
    //取消非活跃超时销毁
    void CancelInactivearease(){
        loop_->RunInLoop(std::bind(&Connection::CancelInactiveareaseInLoop,this));
    }
    //协议切换,就是把会回调函数和上下文切换掉
    void Upgrade(const Any &context,
                const ConnectedCallBack &connected,
                const MessageCallBack &message,
                const ClosedCallBack &closed,
                const AnyEventCallBack &event,
                const SrvClosedCallBack &srvclose)
    {
        loop_->AssertInLoop();
        loop_->RunInLoop(std::bind(&Connection::UpgradeInLoop,this,context,connected,message,closed,event,srvclose));
    }

    std::string ClientIp(){
        const std::string &clientIp = inet_ntoa(client_.sin_addr);
        return clientIp;
    }

    std::uint16_t ClientPort(){
        uint16_t ClientPort = ntohs(client_.sin_port);
        return ClientPort;
    }
};

class Acceptor
{
private:
    EventLoop *loop_;
    Socket socket_;
    Channel channel_;


    using AcceptCallBack = std::function<void(int,struct sockaddr_in)>;
    AcceptCallBack accept_callback_;
private:
    int CreateServer_(uint16_t port)
    {
        socket_.CreateServer(port);
        return socket_.Fd();
    }
public:
    Acceptor(EventLoop *loop,uint16_t port):loop_(loop),channel_(loop_,CreateServer_(port)){
        //绑定监听套接字的可读事件回调
        channel_.SetReadCallback(std::bind(&Acceptor::HandleRead,this));
    }
    void HandleRead(){
        //获取新连接
        struct sockaddr_in client;
        int fd = socket_.Accept(&client);
        if(fd < 0){
            return;
        }
        //调用服务器模块设置的新连接回调
        return accept_callback_(fd,client);
    }
public:
    void SetAcceptCallBack(const AcceptCallBack &cb) { accept_callback_ = cb; }
    void Listen(){
        //启动读事件监控
        channel_.EnableRead();
    }
};



class LoopThread
{
private:
    //用来保证获取EventLoop对象的时候，EventLoop已经被实例化
    std::mutex mutex_;
    std::condition_variable cond_;

    EventLoop *loop_;
    std::thread thread_;
private:
    //线程的入口函数，实例化EventLoop对象，执行EventLoop的监控任务
    void EntryThread()
    {
        //实例化EventLoop对象，这里new在堆上，主要原因是我们想让这个EventLoop对象的生命周期随着LoopThread结束而结束
        EventLoop loop;
        {
            std::unique_lock<std::mutex> lock(mutex_);
            loop_ = &loop;
            //唤醒阻塞在条件变量上的线程
            cond_.notify_all();
        }
        
        loop_->start();
    }
public:
    //创建线程
    LoopThread():loop_(nullptr),thread_(&LoopThread::EntryThread,this) {}
    //获取线程所绑定的EventLoop
    EventLoop *GetLoop(){
        EventLoop *loop = nullptr;
        {
            std::unique_lock<std::mutex>lock (mutex_);
            cond_.wait(lock,[&](){ return loop_ != nullptr; });    //当loop_为nullptr就一直循环重复阻塞
            loop = loop_;
        }
        return loop;
    }
};


class LoopThreadPool
{
private:
    int thread_count_;  //创建的从属线程数量
    int next_index_;    //给连接分配线程的下标
    EventLoop *baseloop_;   //主Reactor的loop
    std::vector<LoopThread *> threads_; //存储创建的从属线程
    std::vector<EventLoop *> loops_;    //存储从属线程绑定的loop

public:
    LoopThreadPool(EventLoop *baseloop) : thread_count_(0), next_index_(0), baseloop_(baseloop) {}
    //设置线程创建数量
    void SetThreadCount(int count) { thread_count_ = count; }
    //创建线程并运行
    void CreatAndStart()
    {
        if (thread_count_ > 0)
        {
            threads_.resize(thread_count_);
            loops_.resize(thread_count_);

            for (int i = 0; i < thread_count_; i++)
            {
                threads_[i] = new LoopThread();
                loops_[i] = threads_[i]->GetLoop();
            }
        }
    }
    //返回一个从属线程对应的loop，采用RR轮转的方式
    EventLoop *NextLoop()
    {
        //如果设置从属线程的数量为0，那么返回主reactor对应的loop,此时服务器就是一个单Reactor单进程的服务器，这个loop既要处理新连接，又要负责连接的监控以及处理
        if (thread_count_ == 0)
        {
            return baseloop_;
        }

        next_index_ = (next_index_ + 1) % thread_count_;
        return loops_[next_index_];
    }
};


class TcpServer
{
private:
    uint64_t next_id_; //连接的唯一ID,也用于定时任务ID
    int port_;
    int timeout_;   //定时任务销毁时间
    bool enable_inactive_release_;  //是否开启非活跃定时销毁
    EventLoop baseloop_;    //主线程的loop
    Acceptor acceptor;  //监听套接字
    LoopThreadPool pool_;   //从属reactor的线程池
public:
    std::unordered_map<uint64_t,PtrConnection> conns_; //管理服务器中的新建连接

    using ConnectedCallBack = std::function<void(const PtrConnection &)>;    
    using MessageCallBack = std::function<void(const PtrConnection &,Buffer *)>; 
    using ClosedCallBack = std::function<void(const PtrConnection &)>;   
    using AnyEventCallBack = std::function<void(const PtrConnection &)>;  
    using Functor = std::function<void()>;
    ConnectedCallBack connected_callback_;              //连接建立回调
    MessageCallBack message_callback_;              //新消息到来回调
    ClosedCallBack closed_callback_;                //连接关闭回调
    AnyEventCallBack event_callback_;               //任意事件回调
private:
    void RunAfterInLoop(const Functor &task,uint32_t delay){
        next_id_++;
        baseloop_.TimerAdd(next_id_,delay,task);
    }
    //acceptor的读事件处理回调
    void NewConnection(int fd,struct sockaddr_in client){
        next_id_++;

        PtrConnection conn(new Connection(pool_.NextLoop(),next_id_,fd,client));
        conn->SetConnectedCallBack(connected_callback_);
        conn->SetMessageCallBack(message_callback_);
        conn->SetClosedCallBack(closed_callback_);
        conn->SetAnyEventCallBack(event_callback_);
        conn->SetSrvClosedCallBack(std::bind(&TcpServer::RemoveConnection,this,std::placeholders::_1));
        if(enable_inactive_release_ == true)
            conn->EnableInactiveRelease(timeout_);
        conn->Established();

        conns_.insert({next_id_,conn});
    }

    void RemoveConnectionInLoop(const PtrConnection &conn){
        int conn_id = conn->Id();

        auto it = conns_.find(conn_id);
        if(it != conns_.end()){
            conns_.erase(it);
        }
        
    }
    //从服务器管理的连接中移除，必须保证在baseloop对应的线程中执行，因为维护服务器连接的conns_在主线程中,所以这里是调用的baseloop的RunInLoop
    void RemoveConnection(const PtrConnection &conn){
        baseloop_.RunInLoop(std::bind(&TcpServer::RemoveConnectionInLoop,this,conn));
    }
public:
    TcpServer(int port)
        :next_id_(0),
        port_(port),
        enable_inactive_release_(false),
        acceptor(&baseloop_,port_),
        pool_(&baseloop_)
    {
        acceptor.SetAcceptCallBack(std::bind(&TcpServer::NewConnection,this,std::placeholders::_1,std::placeholders::_2));
        acceptor.Listen();
    }

    void SetConnectedCallBack(const ConnectedCallBack &cb) { connected_callback_ = cb; }
    void SetMessageCallBack(const MessageCallBack &cb) { message_callback_ = cb; }
    void SetClosedCallBack(const ClosedCallBack &cb) {closed_callback_ = cb; }
    void SetAnyEventCallBack(const AnyEventCallBack &cb) { event_callback_ = cb; }

    //启动非活跃连接超时销毁
    void EnableInactiveRelease(int sec) {
        timeout_ = sec;
        enable_inactive_release_ = true;
    }
    //设置从属线程池的线程数量
    void SetThreadCount(int count) { pool_.SetThreadCount(count); }
    //添加一个定时任务
    void RunAfter(const Functor &task,uint32_t delay){
        baseloop_.RunInLoop(std::bind(&TcpServer::RunAfterInLoop,this,task,delay));
    }
    //启动服务器，就是开启baseloop的读监控
    void Start(){
        pool_.CreatAndStart();
        baseloop_.start();
    }

    //用户实现获取在线人数功能
    int ConnCount(){
        return conns_.size();
    }
};